package org.zn.note.jdk.container;

import org.zn.note.jdk.util.SpringStopWatch;

import java.util.*;
import java.util.concurrent.*;

/**
 * 编译器直接支持的持有对象的方式：1、直接引用 2、数组
 * Java提供容器工具类，同时也是多种数据结构的实现：Java Collections Framework
 * “Java集合类架构” 分为两大类：
 * 1、{@link Collection}  “单值集合”  --> {@link List}、{@link Set}、{@link Queue}
 * 2、{@link Map} "键值对集合"
 * <p>
 * 一、{@link Collection}
 * Collection extends Iterable，所以所有的“单值集合”都是iterable"可迭代"的
 * 关于{@link Iterable}，接口“可迭代的”，只有3个方法：
 * 1、Iterator<T> iterator(); 返回迭代器{@link Iterator}
 * 2、default void forEach(Consumer<? super T> action) {}  forEach的默认实现，支持forEach
 * default Spliterator<T> spliterator() jdk1.8以后加入的，用于Stream并发，可以给集合分段
 * 关于{@link Iterator}迭代器
 * 1、boolean hasNext();
 * 2、E next(); 如果next没有了，则 {@link NoSuchElementException}
 * 3、void remove()
 * 4、default void forEachRemaining(Consumer<? super E> action) {} 默认的forEach实现
 * <p>
 * {@link List} 有序可重复，都可以放null
 * {@link ArrayList} 线程不安全，数组，每次扩容当前的一半。 {@link JdkArrayList}
 * {@link LinkedList} 线程不安全，链表，同时也实现了队列Queue、双端队列Deque。而双端队列本身就可作为栈Stack用 {@link JdkLinkedList}
 * {@link Vector} 线程安全，“向量”，但是读写是一把锁，效率低
 * {@link Stack} 线程安全，继承自Vector
 * {@link CopyOnWriteArrayList} 线程安全，数组，读一样，写时加锁-copy-改-覆盖。读不锁写锁 {@link JdkCopyOnWriteArrayList}
 * 其他线程安全的方法：
 * {@link Collections#synchronizedCollection(Collection)}
 * {@link Collections#synchronizedList(List)}
 * <p>
 * {@link Set} 不重复，都可以放null（除了ConcurrentSkipListSet）
 * {@link HashSet}  通过 HashMap 实现，线程不安全。
 * {@link LinkedHashSet} 通过 LinkedHashMap 实现。继承自HashSet，所有方法都是HashSet的。线程不安全。
 * {@link EnumSet} 自己实现的，只能放{@link Enum}枚举值，内部是一个枚举数组
 * {@link TreeSet} 通过 TreeMap 实现，线程不安全。
 * {@link CopyOnWriteArraySet} 通过 CopyOnWriteArrayList 实现，线程安全
 * {@link ConcurrentSkipListSet} 通过 ConcurrentSkipListMap 实现，线程安全
 * （以上除了“CopyOnWrite...”、“ConcurrentSkip...”是java.util.concurrent，其他全是java.util）
 * <p>
 * {@link Queue} 队列 （双端队列是Deque），都不能放null（除了LinkedList），API{@link #testQueue()}
 * {@link LinkedList} 链表实现的
 * {@link ArrayDeque} 数组，指针实现的环形数组。ArrayDeque性能好。可以学习一下如何用指针实现环形数组，很巧妙。
 * {@link PriorityQueue} 数组，“小顶堆”，log(N)，数组内存储是广度优先的，然后每次poll出“最小的” {@link #testPriorityQueue()}
 * {@link ConcurrentLinkedDeque} 无界Deque，不能null，CAS，不阻塞  {@link #testConcurrentLinkedDeque()}
 * <p>
 * {@link BlockingQueue} 接口，下面都是 BlockingQueue。 取\放可以阻塞、可以timeout  {@link #testBlockingQueue()}
 * {@link LinkedBlockingQueue} 链表，不能扩容（但容量可以是Integer.MAX_VALUE），读写两把锁，非公平锁
 * {@link ArrayBlockingQueue} 数组，不能扩容，读写一把锁！性能不如 LinkedBlockingQueue，但是锁可以设置成公平锁（先到先得）
 * {@link PriorityBlockingQueue} 支持优先级的无界队列，内部是数组，类似PriorityQueue
 * {@link DelayQueue} 内部是 PriorityQueue，队列的元素需要 E extends {@link Delayed}，到时间了才能取出
 * {@link SynchronousQueue} 内部只有一个元素的阻塞队列，插入阻塞等读取，读取阻塞等插入
 * {@link LinkedBlockingDeque} 和LinkedBlockingQueue类似，双向队列
 * {@link LinkedTransferQueue} 1.7新增，推荐使用，详见 {@link #testLinkedTransferQueue()}
 * (LinkedList\ArrayDeque\PriorityQueue，在java.util包，是线程不安全的）
 * （其他Queue都在java.util.concurrent包，都是线程安全的)
 * <p>
 * {@link Map}  （一般涉及红黑树的这种，都是 log(N)）
 * {@link Hashtable} 线程安全，数组{单项链表<K-V>}，KV可null
 * {@link HashMap} --> {@link JdkHashMap}，线程不安全
 * {@link LinkedHashMap} 可以按put顺序，也可以按访问顺序（构造函数指定），默认是按put顺序
 * {@link ConcurrentHashMap} KV都不可null！和HashMap不一样。1.7多段写锁，读不加锁。1.8不用Segment，用CAS + synchronized，增加红黑树
 * {@link ConcurrentSkipListMap} 线程安全的Key排序Hash表
 * {@link TreeMap} 线程不安全的Key排序Hash表
 * {@link WeakHashMap}
 * {@link EnumMap}
 * {@link IdentityHashMap}
 * 场景：
 * 1、单线程、无序：{@link HashMap}
 * 2、单线程、插入顺序：{@link LinkedHashMap}
 * 3、单线程、Key排序：{@link TreeMap}
 * 4、低并发：读写同一把锁 {@link Collections#synchronizedMap(Map)} {@link Collections#synchronizedSortedMap(SortedMap)}
 * 5、高并发、无序：ConcurrentHashMap （不建议Hashtable虽然也是线程安全的）
 * 6、高并发、Key排序：ConcurrentSkipListMap
 */
public class JavaCollectionsFramework {

    // Qeueu API：add\offer、remove\poll、element\peek
    // offer/poll/peek 不抛Exception
    private static void testQueue() {
        Queue<Integer> queue = new LinkedList();
        boolean operateSuccess1 = queue.add(1); // true or IllegalStateException
        boolean operateSuccess2 = queue.offer(2); // true or false
        Integer i1 = queue.remove(); // delete and return element, if empty throw NoSuchElementException
        Integer i2 = queue.poll(); // delete and return element, if empty return null
        Integer i3 = queue.element(); // not delete, if empty throw NoSuchElementException
        Integer i4 = queue.peek(); // not delete, if empty return null
    }

    // 测试性能：ArrayDeque、LinkedList
    // 不考虑线程安全、无优先级：ArrayDeque 比 LinkedList 性能好
    // 不考虑线程安全、有优先级：PriorityQueue
    private static void testArrayDequePerformance() {
        QueuePerformanceTest.testArrayDequeAndLinkedList(); // 测试 ArrayDeque 和 LinkedList 性能
    }

    // 测试性能：BlockingQueue
    // LinkedTransferQueue > LinkedBlockingQueue 远大于 ArrayBlockingQueue
    private static void testBlockingQueuePerformance() {
        int dataNum = 10000000; // 1千万数据
        List<BlockingQueue> list = Arrays.asList(
//                new ArrayBlockingQueue(dataNum),
//                new LinkedBlockingQueue()
                new LinkedTransferQueue()
        );
        SpringStopWatch stopWatch = new SpringStopWatch();
        list.forEach(it -> {
            try {
                QueuePerformanceTest.testBlockQueue(it, stopWatch, dataNum, 100, 100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println(stopWatch.prettyPrint());
    }

    /*
    PriorityQueue
    线程不安全的优先队列，每次poll出“最小”的
    （如果想每次poll出最大的，则需要自己在比较方法中实现，就和正常的反着来）
    优先使用构造函数传入的Comparator，如果没有，则会将 E 强转成 Comparable（没有Comparator则元素必须是Comparable的），否则报错ClassCastException
    通过“二叉小顶堆”实现，所有子节点小于等于父节点，以此类推。
     */
    private static void testPriorityQueue() {
        Queue queue = new PriorityQueue();
        queue.offer(1);
        queue.offer(3);
        queue.offer(2);
        System.out.printf("%d, %d, %d", queue.poll(), queue.poll(), queue.poll());
        queue.offer(new Object());
        queue.offer(new Object());
    }

    private static void testConcurrentLinkedDeque() {
        Queue queue = new ConcurrentLinkedDeque();
        System.out.println(queue.poll());
    }

    // BlockingQueue API
    private static void testBlockingQueue() throws InterruptedException {
        BlockingQueue blockingQueue = new LinkedBlockingDeque(); // 暂时先用这个实现，这里主要是看BlockingQueue API
        // 生产
        blockingQueue.add(new Object()); // 不阻塞，能放返true，满了异常IllegalStateException
        blockingQueue.offer(new Object());  // 不阻塞，能放true，满了false
        blockingQueue.put(new Object()); // 返void，阻塞wait，如果wait被中断则InterruptedException
        blockingQueue.offer(new Object(), 1L, TimeUnit.SECONDS); // 阻塞，指定wait时间，InterruptedException
        // 消费
        blockingQueue.take(); // 阻塞wait，InterruptedException
        blockingQueue.poll(1L, TimeUnit.SECONDS); // 阻塞，指定wait时间，InterruptedException
    }

    /*
    LinkedTransferQueue<E> implements 【TransferQueue<E> extends BlockingQueue<E>】   “transfer” 转让、转移
    1.7 新增的接口 TransferQueue，目前只有一个实现 LinkedTransferQueue，性能比LinkedBlockingQueue好
    是ConcurrentLinkedQueue、SynchronousQueue、LinkedBlockingQueue的超集，更好。
     */
    private static void testLinkedTransferQueue() throws InterruptedException {
        Object element = new Object();
        TransferQueue q = new LinkedTransferQueue();
        // 不阻塞、立即返回：当前有“waiting consumer”返true，说明有消费者接收了。否则返false，element不存直接扔了。
        boolean transferOK = q.tryTransfer(element); // Transfers the element to a waiting consumer immediately
        // 阻塞，直到有“waiting consumer”来接收。返void。阻塞中断会 InterruptedException （所有阻塞wait中断都InterruptedException）
        q.transfer(element);
        // 阻塞，就是tryTransfer加了超时时间
        transferOK = q.tryTransfer(element, 1L, TimeUnit.SECONDS);
        // Returns true if there is at least one consumer waiting to receive an element
        boolean hasWaitingConsumer = q.hasWaitingConsumer();
        // Returns an estimate of the number of consumers waiting to receive elements
        int waitingConsumerCount = q.getWaitingConsumerCount(); // "估计"的消费者数，因为是某一时刻的，随时会变
    }

    static class QueuePerformanceTest {
        /*
        测试 ArrayDeque 和 LinkedList 性能： 综合来看 ArrayDeque 性能好
        1千万数据：  ArrayDeque   LinkedList  （单位：毫秒ms）
        offerFirst       4519         3655
        pollLast           15          130
        offerLast         597         7439
        pollFirst          16          142
         */
        static void testArrayDequeAndLinkedList() {
            List<Deque<Integer>> deques1 = Arrays.asList(new ArrayDeque(), new LinkedList());
            SpringStopWatch stopWatch = new SpringStopWatch();
            int times = 10000000; // 1千万数据
            deques1.forEach(it -> {
                stopWatch.start(it.getClass().getSimpleName() + ".offerFirst");
                for (int i = 0; i < times; i++)
                    it.offerFirst(i);
                stopWatch.stop();
                stopWatch.start(it.getClass().getSimpleName() + ".pollLast");
                for (int i = 0; i < times; i++)
                    it.pollLast();
                stopWatch.stop();
            });
            List<Deque<Integer>> deques2 = Arrays.asList(new ArrayDeque(), new LinkedList());
            deques2.forEach(it -> {
                stopWatch.start(it.getClass().getSimpleName() + ".offerLast");
                for (int i = 0; i < times; i++)
                    it.offerLast(i);
                stopWatch.stop();
                stopWatch.start(it.getClass().getSimpleName() + ".pollFirst");
                for (int i = 0; i < times; i++)
                    it.pollFirst();
                stopWatch.stop();
            });
            System.out.println(stopWatch.prettyPrint());
        }

        /*
        性能测试：BlockingQueue
        LinkedTransferQueue > LinkedBlockingQueue >> ArrayBlockingQueue
        LinkedTransferQueue 更快更省内存，但是CAS可能更耗CPU
        1、LinkedBlockingQueue，读写两把(非公平)锁，每把锁newCondition，满了或空了会await，好了就signal，性能损耗是ReentrantLock
        2、LinkedTransferQueue，用的是CAS。
        消费者如果没数据，也创建一个节点，然后卡在这个节点上。（预占）
        生产者来了直接把数据放到这个节点上、不用入队了，然后唤醒这个节点的消费者。（匹配）
         */
        static void testBlockQueue(BlockingQueue q, SpringStopWatch stopWatch, int dataNum,
                                   int productorThreadNum, int consumerThreadNum) throws InterruptedException {
            Object o = new Object();
            CountDownLatch countDownLatch = new CountDownLatch(dataNum);
            ExecutorService productors = Executors.newFixedThreadPool(productorThreadNum);
            ExecutorService consumers = Executors.newFixedThreadPool(consumerThreadNum);
            stopWatch.start(String.format("%s, dataNum[%d], productorThreadNum[%d], consumerThreadNum[%d]",
                    q.getClass().getSimpleName(), dataNum, productorThreadNum, consumerThreadNum));
            for (int i = 0; i < dataNum; i++) {
                productors.submit(() -> {
                    try {
                        q.put(o);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
                consumers.submit(() -> {
                    try {
                        q.take();
                        countDownLatch.countDown();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            }
            countDownLatch.await();
            stopWatch.stop();
            productors.shutdown();
            consumers.shutdown();
        }
    }

    public static void main(String[] args) {
        testBlockingQueuePerformance();
    }
}